package com.example.kafka2.controller;


import com.example.kafka2.model.Product;
import com.example.kafka2.model.ProductSource;
import com.example.kafka2.util.KafkaUtils;
import com.google.gson.Gson;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
//多线程
@RestController
@RequestMapping("/pocyb")
public class ProductYbController {

    private static boolean isrun2 = false;
    private static AtomicInteger incr2 = new AtomicInteger(0);
    @RequestMapping("start")
    public Boolean start(Integer num,Integer bfs,String topica,String topicb,Integer sleep,Integer fl,Integer pro,Integer sou){
        try {
            //KafkaUtils.init();
            incr2.set(num);
            ExecutorService executorService = Executors.newFixedThreadPool(bfs);
            for (int i = 1; i <= bfs; i++) {
                final int m = i;
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            extracted(m,topica, topicb, sleep, fl,pro, sou);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
            executorService.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    private void extracted(int m,String topica, String topicb, Integer sleep,Integer fl,Integer pro,Integer sou) throws InterruptedException, ExecutionException {
        isrun2 = true;
        Random rand = new Random();
        Gson gson = new Gson();
        while (isrun2) {
            if (sleep != null && sleep > 0) {
                Thread.sleep(sleep);
            }
            int incrementAndGet = incr2.incrementAndGet();
           // System.out.println("xiancheng"+m+"[[[[[[[[[[[[[[[[["+incrementAndGet+"]]]]]]]]]]]]]]]]"+System.currentTimeMillis());
            int merchant_fullname = rand.nextInt(fl)+1;  // 获取一个随机整数
            int product_id = rand.nextInt(pro)+1;  // 获取一个随机整数
            int score = rand.nextInt(51)+50;
            long ts =System.currentTimeMillis();
            Product pdt = new Product("分类"+merchant_fullname,"产品"+product_id,
                    "product"+product_id,"v1.1",incrementAndGet,
                    score,""+product_id,ts);
            int soure_id = rand.nextInt(sou)+1;  // 获取一个随机整数
            ProductSource productSource = new ProductSource(incrementAndGet,""+soure_id,ts);
            KafkaUtils.sendybmsg(topica, gson.toJson(pdt),incrementAndGet);
            KafkaUtils.sendybmsg(topicb, gson.toJson(productSource),incrementAndGet);
        }
    }

    @RequestMapping("stop")
    public Boolean stop(){
        isrun2=false;
        //KafkaUtils.producer.close();
        return isrun2;
    }
}
